package io.gitee.hefren.hhmq.server;

import com.google.common.base.MoreObjects;
import static com.google.common.base.Preconditions.*;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import io.gitee.hefren.hhmq.model.HhMessage;
import io.gitee.hefren.hhmq.model.HhConsumerSubscription;
import io.gitee.hefren.hhmq.store.HhIndexer;
import io.gitee.hefren.hhmq.store.HhStore;

import java.util.List;
import java.util.Map;
import java.util.Objects;


/**
 * @Date 2024/7/23
 * @Author lifei
 */
public class HhMQ {

    private static final Map<String, HhMQ> topicMQMap = Maps.newHashMap();
    private static final Map<String, HhConsumerSubscription> consumerTopicMap = Maps.newHashMap();
    private static final String DEFAULT_TOPIC = "com.hh.demo";
    static {
        topicMQMap.put(DEFAULT_TOPIC, new HhMQ(DEFAULT_TOPIC));
    }

    private String topic;
    private HhStore hhStore;

    public HhMQ(String topic) {
        this.topic = topic;
        this.hhStore = new HhStore(topic);
        this.hhStore.init();
    }

    public static Integer sendMessage(String topic, HhMessage<?> hhMessage) {
        HhMQ hhMQ = topicMQMap.get(topic);
        checkState(Objects.nonNull(hhMQ), "Topic not exists, Please create topic first: %s", topic);
        return hhMQ.send(hhMessage);
    }

    public static void subscribeTopic(String topic, String consumerId) {
        HhMQ hhMQ = topicMQMap.get(topic);
        checkState(Objects.nonNull(hhMQ), "Topic not exists: %s", topic);
        hhMQ.subscribe(topic, consumerId);
    }

    public static void unsubscribeTopic(String topic, String consumerId) {
        HhMQ hhMQ = topicMQMap.get(topic);
        checkState(Objects.nonNull(hhMQ), "Topic not exists: %s", topic);
        hhMQ.unsubscribe(consumerId);
    }

    public static List<HhMessage<?>> batchReceiveMessage(String topic, String consumerId, Integer size) {
        HhMQ hhMQ = topicMQMap.get(topic);
        checkState(Objects.nonNull(hhMQ), "Topic not exists: %s", topic);
        HhConsumerSubscription hhConsumerSubscription = consumerTopicMap.get(consumerId);
        checkState(Objects.nonNull(hhConsumerSubscription), "Please subscribe topic first: %s", topic);
        List<HhMessage<?>> result = Lists.newArrayList();
        int ackOffset = hhConsumerSubscription.getOffset();
        for (int i=0; i<size; i++) {
            int offset = 0;
            if (ackOffset!=-1) {
                HhIndexer.Entry entry = HhIndexer.getEntry(topic, ackOffset);
                if (Objects.isNull(entry)) {
                    break;
                }
                offset = ackOffset + entry.getLen();
            }
            HhMessage<?> hhMessage = receiveMessage(topic, consumerId, offset);
            if (Objects.isNull(hhMessage)) {
                break;
            }
            result.add(hhMessage);
            ackOffset = offset;
        }
        return result;
    }

    /**
     * 调用这个方法，还需要手动调用ack方法，移动偏移量
     * @param topic
     * @param consumerId
     * @return
     */
    public static HhMessage<?> receiveMessage(String topic, String consumerId) {
        HhMQ hhMQ = topicMQMap.get(topic);
        checkState(Objects.nonNull(hhMQ), "Topic not exists: %s", topic);
        HhConsumerSubscription hhConsumerSubscription = consumerTopicMap.get(consumerId);
        checkState(Objects.nonNull(hhConsumerSubscription),"Please subscribe topic first: %s", topic);
        // 最新确认的偏移量
        int ackOffset = hhConsumerSubscription.getOffset();
        // 如果是第一次消息
        int offset = 0;
        // 如果不是第一次消息
        if (ackOffset!=-1) {
            // 最新消费的这条消息的长度
            HhIndexer.Entry entry = HhIndexer.getEntry(topic, ackOffset);
            if (Objects.isNull(entry)) {
                return null;
            }
            offset = ackOffset + entry.getLen();
        }
        return receiveMessage(topic, consumerId, offset);
    }

    public static HhMessage<?> receiveMessage(String topic, String consumerId, int offset) {
        HhMQ hhMQ = topicMQMap.get(topic);
        checkState(Objects.nonNull(hhMQ), "Topic not exists: %s", topic);
        HhConsumerSubscription hhConsumerSubscription = consumerTopicMap.get(consumerId);
        checkState(Objects.nonNull(hhConsumerSubscription), "Please subscribe topic first: %s", topic);
        return hhMQ.receive(offset);
    }

    public static Integer ackMessage(String topic, String consumerId, Integer offset) {
        HhMQ hhMQ = topicMQMap.get(topic);
        checkState(Objects.nonNull(hhMQ), "Topic not exists: %s", topic);
        HhConsumerSubscription hhConsumerSubscription = consumerTopicMap.get(consumerId);
        checkState(Objects.nonNull(hhConsumerSubscription), "Please subscribe topic first: %s", topic);
        return hhMQ.ack(hhConsumerSubscription, offset);
    }



    private Integer ack(HhConsumerSubscription hhConsumerSubscription, Integer offset) {
        if (offset>hhConsumerSubscription.getOffset() && offset< HhStore.LEN) {
            hhConsumerSubscription.setOffset(offset);
            return offset;
        }
        return -1;
    }

    private HhMessage<?> receive(int offset) {
        return hhStore.read(offset);
    }

    private void unsubscribe(String consumerId) {
        consumerTopicMap.remove(consumerId);
    }

    private void subscribe(String topic, String consumerId) {
        consumerTopicMap.putIfAbsent(consumerId, new HhConsumerSubscription(consumerId, topic, -1));
    }

    private Integer send(HhMessage<?> hhMessage) {
        if (Objects.isNull(hhMessage)) {
            return -1;
        }
        // 获取消息的偏移量
        int offset = hhStore.getMessageOffset();
        hhMessage.getHeaders().put(HhMessage.OFFSET_KEY, String.valueOf(offset));
        hhStore.write(hhMessage);
        return offset;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }


    @Override
    public String toString() {
        return MoreObjects.toStringHelper(HhMQ.class)
                .add("topic", topic)
                .toString();
    }
}
